﻿#region Copyright Notice
// 
// 
//  
//  Copyright (c) 2009-2011 Anindya Chatterjee
//  
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//  
//      http://www.apache.org/licenses/LICENSE-2.0
//  
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// 
// 
#endregion
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using AbstractClass.Collections.Generic;
using AbstractClass.Extensions;
using AbstractClass.Linq;

namespace AbstractClass.Threading.Tasks
{
    /// <summary>
    /// Represents a type which executes and monitors the operation of a number of scheduled
    /// <see cref="Job&lt;TResult&gt;"/>.
    /// </summary>
    /// <typeparam name="TResult">The type of the result of the operation.</typeparam>
    public class JobEngine<TResult>
    {
        private readonly List<KeyValuePair<string, TResult>> _results;
        private bool _cancelled;
        private bool _started;

        /// <summary>
        /// Initializes a new instance of the <see cref="JobEngine&lt;TResult&gt;"/> class.
        /// </summary>
        public JobEngine()
        {
            JobQueue = new RandomAccessQueue<List<Job<TResult>>>();
            _results = new List<KeyValuePair<string, TResult>>();
        }

        internal RandomAccessQueue<List<Job<TResult>>> JobQueue { get; set; }

        /// <summary>
        /// Gets the total number of scheduled jobs.
        /// </summary>
        /// <value>The total number of scheduled jobs.</value>
        public int Size
        {
            get
            {
                int count = JobQueue.Sum(item => item.Count);
                return count;
            }
        }

        /// <summary>
        /// Occurs when one sequential job or a number of parallel jobs complete execution.
        /// </summary>
        public event EventHandler<ProgressChangedEventArgs> ProgressChanged;

        /// <summary>
        /// Occurs when one job completes its execution.
        /// </summary>
        public event EventHandler<JobCompletedEventArgs<TResult>> JobCompleted;

        /// <summary>
        /// Occurs when the engine starts execution.
        /// </summary>
        public event EventHandler ExecutionStarted;

        /// <summary>
        /// Occurs when the engine completes execution of all scheduled jobs.
        /// </summary>
        public event EventHandler ExecutionEnded;

        /// <summary>
        /// Occurs when engine cancels execution of all jobs.
        /// </summary>
        public event EventHandler ExecutionCancelled;

        /// <summary>
        /// Schedules all jobs queued in <paramref name="engine"/> to be run before the execution of <paramref name="job"/>.
        /// </summary>
        /// <param name="engine">A <see cref="JobEngine&lt;TResult&gt;"/> instance containing other scheduled jobs.</param>
        /// <param name="job">The last job to run.</param>
        /// <returns>A <see cref="JobEngine&lt;TResult&gt;"/> instance to start and monitor the execution.</returns>
        /// <exception cref="InvalidOperationException">Job scheduling fails.</exception>
        public static JobEngine<TResult> operator +(JobEngine<TResult> engine, Job<TResult> job)
        {
            job.ThrowIfNull("job");

            try
            {
                if (engine == null)
                    engine = new JobEngine<TResult>();
                engine.Enqueue(job);
                return engine;
            }
            catch (Exception ex)
            {
                throw new InvalidOperationException(LanguageResource.JobSequenceOperationFailed, ex);
            }
        }

        /// <summary>
        /// Schedules last jobs queued in <paramref name="engine"/> to be run parallely with <paramref name="job"/>.
        /// </summary>
        /// <param name="engine">A <see cref="JobEngine&lt;TResult&gt;"/> instance containing other scheduled jobs.</param>
        /// <param name="job">The last job to run parallely.</param>
        /// <returns>A <see cref="JobEngine&lt;TResult&gt;"/> instance to start and monitor the execution.</returns>
        /// <exception cref="InvalidOperationException">Job scheduling fails.</exception>
        public static JobEngine<TResult> operator -(JobEngine<TResult> engine, Job<TResult> job)
        {
            job.ThrowIfNull("job");

            try
            {
                var engineA = new JobEngine<TResult>();
                engineA.Enqueue(job);
                return engine - job.Engine;
            }
            catch (Exception ex)
            {
                throw new InvalidOperationException(LanguageResource.JobParallelOperationFailed, ex);
            }
        }

        /// <summary>
        /// Schedules jobs in such a way that the last job queued in <paramref name="engineA"/> would run parallely with the
        /// first job in <paramref name="engineB"/>.
        /// </summary>
        /// <param name="engineA">A <see cref="JobEngine&lt;TResult&gt;"/> instance containing scheduled jobs.</param>
        /// <param name="engineB">A <see cref="JobEngine&lt;TResult&gt;"/> instance containing other scheduled jobs.</param>
        /// <returns>A <see cref="JobEngine&lt;TResult&gt;"/> instance to start and monitor the execution.</returns>
        /// <exception cref="InvalidOperationException">Job scheduling fails.</exception>
        public static JobEngine<TResult> operator -(JobEngine<TResult> engineA, JobEngine<TResult> engineB)
        {
            engineA.ThrowIfNull("engineA");
            engineB.ThrowIfNull("engineB");

            try
            {
                List<Job<TResult>> tail = engineA.JobQueue.Tail();
                List<Job<TResult>> head = engineB.JobQueue.Peek();

                if (tail == null)
                {
                    engineA.JobQueue.Enqueue(new List<Job<TResult>>());
                    tail = engineA.JobQueue.Tail();
                }

                if (head == null)
                {
                    engineB.JobQueue.Enqueue(new List<Job<TResult>>());
                    head = engineB.JobQueue.Peek();
                }

                head.Process(task => task.Engine = engineA);
                tail.AddRange(head);
                foreach (var element in engineB.JobQueue.Where(list => !list.Equals(head)))
                {
                    engineA.EnqueueRange(element);
                }
                return engineA;
            }
            catch (Exception ex)
            {
                throw new InvalidOperationException(LanguageResource.JobParallelOperationFailed, ex);
            }
        }

        /// <summary>
        /// Schedules jobs in such a way that after completion of all jobs queued in <paramref name="engineA"/> the jobs queued
        /// in <paramref name="engineB"/> would start to execute.
        /// </summary>
        /// <param name="engineA">A <see cref="JobEngine&lt;TResult&gt;"/> instance containing scheduled jobs.</param>
        /// <param name="engineB">A <see cref="JobEngine&lt;TResult&gt;"/> instance containing other scheduled jobs.</param>
        /// <returns>A <see cref="JobEngine&lt;TResult&gt;"/> instance to start and monitor the execution.</returns>
        /// <exception cref="InvalidOperationException">Job scheduling fails.</exception>
        public static JobEngine<TResult> operator +(JobEngine<TResult> engineA, JobEngine<TResult> engineB)
        {
            engineA.ThrowIfNull("engineA");
            engineB.ThrowIfNull("engineB");

            try
            {
                foreach (var element in engineB.JobQueue)
                {
                    engineA.EnqueueRange(element);
                }
                return engineA;
            }
            catch (Exception ex)
            {
                throw new InvalidOperationException(LanguageResource.JobSequenceOperationFailed, ex);
            }
        }

        /// <summary>
        /// Schedules a specified job at the end of the internal job queue.
        /// </summary>
        /// <param name="job">The job to schedule.</param>
        public void Enqueue(Job<TResult> job)
        {
            job.ThrowIfNull("job");

            JobQueue.Enqueue(new List<Job<TResult>> {job});
            job.Engine = this;
        }

        /// <summary>
        /// Schedules a number of jobs sequentially at the end of the internal job queue.
        /// </summary>
        /// <param name="jobs">The sequence of jobs to schedule.</param>
        public void EnqueueRange(IEnumerable<Job<TResult>> jobs)
        {
            var list = new List<Job<TResult>>(jobs);
            list.Process(task => task.Engine = this);
            JobQueue.Enqueue(list);
        }

        /// <summary>
        /// Schedules <paramref name="newJob"/> to run after completion of <paramref name="existingJob"/>.
        /// </summary>
        /// <param name="existingJob">The existing job.</param>
        /// <param name="newJob">The new job to schedule.</param>
        public void EnqueueSequential(Job<TResult> existingJob, Job<TResult> newJob)
        {
            existingJob.ThrowIfNull("existingJob");
            newJob.ThrowIfNull("newJob");

            foreach (var item in JobQueue.Where(item => item.Contains(existingJob)))
            {
                JobQueue.AddAfter(item, new List<Job<TResult>> {newJob});
                newJob.Engine = this;
                return;
            }

            var tasks = new List<Job<TResult>> {existingJob};
            JobQueue.Enqueue(tasks);
            existingJob.Engine = this;
            JobQueue.AddAfter(tasks, new List<Job<TResult>> {newJob});
            newJob.Engine = this;
        }

        /// <summary>
        /// Schedules <paramref name="newJob"/> to run parallely with <paramref name="existingJob"/>.
        /// </summary>
        /// <param name="existingJob">The existing job.</param>
        /// <param name="newJob">The new job to schedule.</param>
        public void EnqueueParallel(Job<TResult> existingJob, Job<TResult> newJob)
        {
            existingJob.ThrowIfNull("existingJob");
            newJob.ThrowIfNull("newJob");

            foreach (var item in JobQueue.Where(item => item.Contains(existingJob)))
            {
                item.Add(newJob);
                newJob.Engine = this;
                return;
            }
            JobQueue.Enqueue(new List<Job<TResult>> {existingJob, newJob});
            newJob.Engine = existingJob.Engine = this;
        }

        /// <summary>
        /// Starts the execution of all scheduled jobs.
        /// </summary>
        public void Start()
        {
            if (!_started)
            {
                _started = true;
                _cancelled = false;
            }
            else return;

            EventHandler started = ExecutionStarted;
            if (started != null)
                started(this, new EventArgs());

            int length = JobQueue.Count;
            int count = 0;

            while (JobQueue.Count != 0)
            {
                if (_cancelled)
                {
                    EventHandler cancelled = ExecutionCancelled;
                    if (cancelled != null)
                        cancelled(this, new EventArgs());
                    return;
                }

                List<Job<TResult>> item = JobQueue.Dequeue();
                count++;
                if (item.Count > 1)
                    RunParallely(item);
                else
                    Run(item.SingleOrDefault());

                EventHandler<ProgressChangedEventArgs> progress = ProgressChanged;
                if (progress != null)
                    progress(this, new ProgressChangedEventArgs((100 * count) / length));
            }
            _started = false;

            EventHandler ended = ExecutionEnded;
            if (ended != null)
                ended(this, new EventArgs());
        }

        /// <summary>
        /// Cancels the execution of all scheduled jobs.
        /// </summary>
        public void Cancel()
        {
            _cancelled = true;
        }

        private void Run(Job<TResult> job)
        {
            if (job.Processed)
                throw new InvalidJobOperationException(
                    string.Format(LanguageResource.JobEngineJobAlreadyProcessed, job.JobId));

            TResult result = job.Invoke();
            _results.Add(new KeyValuePair<string, TResult>(job.JobId, result));
            EventHandler<JobCompletedEventArgs<TResult>> completed = JobCompleted;
            if (completed != null)
                completed(this, new JobCompletedEventArgs<TResult>(job, result));
            job.Processed = true;
        }

        private void RunParallely(ICollection<Job<TResult>> tasks)
        {
            ThreadPool.SetMinThreads(tasks.Count, tasks.Count);
            IEnumerable<IAsyncResult> asyncContext = RunAsync(tasks);
            WaitForAll(tasks, asyncContext);
        }

        private static IEnumerable<IAsyncResult> RunAsync(IEnumerable<Job<TResult>> tasks)
        {
            var list = new List<IAsyncResult>();
            foreach (var task in tasks)
            {
                if (task.Processed)
                    throw new InvalidJobOperationException(
                        string.Format(LanguageResource.JobEngineJobAlreadyProcessed, task.JobId));

                list.Add(task.BeginInvoke(null, null));
            }
            return list;
        }

        private void WaitForAll(IEnumerable<Job<TResult>> tasks, IEnumerable<IAsyncResult> asyncContext)
        {
            IEnumerator<IAsyncResult> iterator = asyncContext.GetEnumerator();
            foreach (var assignment in tasks)
            {
                iterator.MoveNext();
                TResult result = assignment.EndInvoke(iterator.Current);
                _results.Add(new KeyValuePair<string, TResult>(assignment.JobId, result));
                EventHandler<JobCompletedEventArgs<TResult>> completed = JobCompleted;
                if (completed != null)
                    completed(this, new JobCompletedEventArgs<TResult>(assignment, result));
                assignment.Processed = true;
            }
        }

        /// <summary>
        /// Gets the results of each job along with its unique id.
        /// </summary>
        /// <returns>A <see cref="Dictionary&lt;String, TResult&gt;"/> containing the result
        /// of each job along with its unique id.</returns>
        public Dictionary<string, TResult> GetResults()
        {
            return _results.ToDictionary(result => result.Key, result => result.Value);
        }
    }
}